package com.ruyuan.little.project.rocketmq.api.order.producer;

import com.ruyuan.little.project.rocketmq.api.order.listener.FinishedOrderTransactionListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 * 订单消息生产者配置
 * @author 强军
 */
@Configuration
public class OrderProducerConfiguration {

    @Value("${rocketmq.namesrv.address}")
    private String namesrvAddress;

    @Value("${rocketmq.order.producer.group}")
    private String orderProducerGroup;

    /**
     * 退房消息通道
     */
    @Value("${rocketmq.order.finished.producer.group}")
    private String orderProducerFinishedGroup;



    @Bean(value = "orderMqProducer")
    public DefaultMQProducer orderMqProducer() throws MQClientException {
        DefaultMQProducer orderMqProducer = new DefaultMQProducer(orderProducerGroup);
        orderMqProducer.setNamesrvAddr(namesrvAddress);
        orderMqProducer.start();
        return orderMqProducer;
    }

    /**
     * 订单事务生产者
     *
     * @param finishedOrderTransactionListener
     * @return
     * @throws MQClientException
     */
    @Bean(value = "orderFinishedTransactionMqProducer")
    public TransactionMQProducer orderTransactionMqProducer(@Qualifier(value = "finishedOrderTransactionListener")
                                                                    FinishedOrderTransactionListener finishedOrderTransactionListener) throws MQClientException {
        TransactionMQProducer transactionMqProducer = new TransactionMQProducer(orderProducerFinishedGroup);
        transactionMqProducer.setNamesrvAddr(namesrvAddress);

        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        transactionMqProducer.setExecutorService(executorService);
        transactionMqProducer.setTransactionListener(finishedOrderTransactionListener);
        transactionMqProducer.start();
        return transactionMqProducer;
    }


}
